package com.atguigu.flink.chapter12;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/11/2 09:38
 */
public class FlinkSQLTopN {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        //1. 先建立一个动态表与数据源(文件)关联
        tEnv.executeSql("create table ub(" +
                            "   user_id bigint, " +
                            "   item_id bigint, " +
                            "   c3_id bigint, " +
                            "   behavior string, " +
                            "   ts bigint, " +
                            "   et as to_timestamp_ltz(ts, 0), " +
                            "   watermark for et as et - interval '3' second " +
                            ")with(" +
                            " 'connector'='filesystem', " +
                            " 'path'='input/UserBehavior.csv', " +
                            " 'format'='csv'" +
                            ")");
        
        // 2. 分组集合: 统计每个商品在每个窗口的点击量
        Table t1 = tEnv.sqlQuery("select " +
                                     " window_start stt, " +
                                     " window_end edt, " +
                                     " item_id, " +
                                     " count(*) ct " +
                                     "from table( tumble(table ub, descriptor(et), interval '2' hour)) " +
                                     "where behavior='pv' " +
                                     "group by window_start, window_end, item_id");
        tEnv.createTemporaryView("t1", t1);
        
        // 3. over 窗口: 给每个窗内内的商品按照点击降序排列, 取名次  rn
        // rank dense_rank row_number(flink sql 只支持这个)
        Table t2 = tEnv.sqlQuery("select " +
                                     " stt, " +
                                     " item_id, " +
                                     " ct, " +
                                     " row_number() over(partition by stt order by ct desc) rn " +
                                     "from t1");
        tEnv.createTemporaryView("t2", t2);
        // 4. 取出 top3
        Table result = tEnv.sqlQuery("select " +
                                         "stt," +
                                         "item_id," +
                                         "ct," +
                                         "rn " +
                                         "from t2 " +
                                         "where rn<=3");
        
        // 5. 把结果输出到 mysql 中. 把 topN 的结果,输出到支持 update 的数据库中
        tEnv.executeSql("CREATE TABLE `abc` (" +
                            "  `stt` timestamp ," +
                            "  `item_id` bigint," +
                            "  `item_count` bigint," +
                            "  `rn` bigint," +
                            "  PRIMARY KEY (`stt`,`rn`) not enforced" +
                            ")with(" +
                            "  'connector' = 'jdbc'," +
                            "   'url' = 'jdbc:mysql://hadoop162:3306/flink_sql?useSSL=false'," +
                            "   'table-name' = 'hot_item', " + // mysql 中的表名
                            "   'username' = 'root', " +
                            "   'password' = 'aaaaaa' " +
                            ")");
        result.executeInsert("abc");
//        tEnv.executeSql("insert into abc select * from " + result);
    }
}
